package main;

import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import worker.SplitWorker;

/**
 * 
 * Provides a set of methods that you can use to partition the graph.
 * 
 */
public class Partitioner {

	/**
	 * Creates initial partition for the graph. (sequential)
	 * 
	 * @param mapGraph
	 *            graph representation.
	 * @return initial partition.
	 */
	public Partition createInitialPartitionSeq(Map<Integer, Node> mapGraph) {

		Map<Integer, Block> mapBlock = new HashMap<Integer, Block>();

		// initial partition is created according to capacity.
		for (Node node : mapGraph.values()) {
			if (!mapBlock.containsKey(node.getCapacity())) {
				mapBlock.put(node.getCapacity(), new Block(node));
			} else {
				mapBlock.get(node.getCapacity()).addNode(node);
			}
		}

		return new Partition(new LinkedList<Block>(mapBlock.values()));
	}

	/**
	 * Parallel post-block partitioning algorithm.
	 * 
	 * @param partition
	 */
	public void partitionPar(Partition partition) {

		// First we need a thread-pool, that would execute threads on each
		// unstable block.
		ThreadPoolExecutor splitWorkerPool = new ThreadPoolExecutor(
				Simulate.splitWorkerPoolCore, Simulate.splitWorkerPoolMax,
				1000, TimeUnit.MILLISECONDS,
				new LinkedBlockingQueue<Runnable>());

		// We count, how many tasks are currently submitted. This number can
		// grow and
		// shrink, since some processed and stable blocks may become unstable
		// again.
		Semaphore numberTasks = new Semaphore(0);
		// We count how many results we got so far. We get some result if the
		// thread is done with its computations and releases one ticket.
		Semaphore numberResults = new Semaphore(0, true);

		do {

			// our split-block that we want to split.
			Block splitBlock = null;

			// while there are some unfixed blocks, we try to split them.
			while ((splitBlock = partition.getUnstableBlock()) != null) {

				// well, is splitBlock is unfixed, release one ticket. One task
				// is now processed.
				numberTasks.release();

				// submit the task
				splitWorkerPool.execute(new SplitWorker(splitBlock, partition,
						numberResults));
				// clean the pointer for the next task
				splitBlock = null;
			}

			try {

				// Count down the number of tasks, that are currently processed
				// if any.
				if (numberTasks.tryAcquire()) {
					// Wait for one result.
					numberResults.acquire();
				}
			} catch (InterruptedException ex) {
				ex.printStackTrace();
			}

			// if we still have some unstable or taken blocks then try again to
			// submit some more tasks,
		} while (partition.hasUnstableOrTakenBlock());

	}

	/**
	 * Sequential Post-Block partitioning algorithm.
	 * 
	 * @param partition
	 */
	public void partitionSeq(Partition partition) {

		while (partition.hasUnstableOrTakenBlock()) {

			Block splitBlock = partition.getUnstableBlockSeq();

			List<Block> postBlocks = partition.getPostBlocks(splitBlock);

			boolean splitted = false;

			for (Block postBlock : postBlocks) {
				Block preset = postBlock.pre();
				Block intersect = preset.intersect(splitBlock);
				Block difference = splitBlock.difference(intersect);

				boolean condition = !difference.isEmpty()
						&& !intersect.isEmpty();

				if (condition) {
					partition.split(splitBlock, intersect, difference);
					splitted = true;
					break;
				}
			}

			if (!splitted) {
				splitBlock.setIrrelevantPostBlocks(postBlocks);
				splitBlock.setStable(true);
			}
		}
	}

	/**
	 * Sequential algorithm that follows the Definition. This algo is slow and
	 * is not used.
	 * 
	 * @param partition
	 */
	public void partitionSeqDefinition(Partition partition) {

		while (partition.hasUnstableOrTakenBlock()) {
			Block splitBlock = partition.getUnstableBlockSeq();
			boolean split = false;
			SortedSet<Node> nodeList = splitBlock.getNodeSet();
			for (Node refNode : nodeList) {
				if (refNode.getPostNodeSet().size() <= 0)
					continue;

				SortedSet<Node> postList4RefNode = refNode.getPostNodeSet();
				SortedSet<Node> listSameBlock = new TreeSet<Node>();
				listSameBlock.add(refNode);
				SortedSet<Node> listOtherBlock = new TreeSet<Node>();
				for (Node postRefNode : postList4RefNode) {
					listSameBlock = new TreeSet<Node>();
					listSameBlock.add(refNode);
					listOtherBlock = new TreeSet<Node>();
					for (Node otherNode : nodeList) {
						if (otherNode == refNode)
							continue;
						boolean added = false;
						for (Node postOtherNode : otherNode.getPostNodeSet()) {
							if (postOtherNode.getBlock() == postRefNode
									.getBlock()) {
								listSameBlock.add(otherNode);
								added = true;
								break;
							}
						}
						if (!added && !listOtherBlock.contains(otherNode)) {
							listOtherBlock.add(otherNode);
						}
					}

					if (!listOtherBlock.isEmpty()) {

						split = true;
						break;
					}
				}

				if (split) {
					for (Node node : nodeList) {
						for (Node preNode : node.getPredNodeSet()) {
							preNode.getBlock().destable();
						}
					}

					partition.split(splitBlock, new Block(listSameBlock, true),
							new Block(listOtherBlock, true));

					break;
				}
			}

			if (!split) {
				splitBlock.setStable(true);
			}

		}
	}

}
